package belf.migrate.engine.job.structure.helper;

import belf.migrate.api.model.QualifiersModel;
import belf.migrate.api.table.schema.*;
import belf.migrate.api.taskconf.JobContext;
import belf.migrate.api.type.DbObjectType;
import belf.migrate.api.type.LogLevelType;
import belf.migrate.api.util.*;
import belf.migrate.engine.job.JobResult;
import belf.migrate.api.job.Job;
import belf.migrate.engine.job.JobResultItem;
import belf.migrate.api.util.ExceptionUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.time.DateFormatUtils;

import java.util.Date;
import java.util.List;

import static belf.migrate.api.job.JobCode.*;

@Slf4j
public class CreateIndexTool extends TableTool {

    public CreateIndexTool(JobContext jobContext, Job job) {
        super(jobContext, job);
        this.job = job;
    }

    @Override
    public JobResult run() {
        PostLog.sendLogModel(
                job.getJobId(),
                job.getTask().getTaskType().getName(),
                "创建索引开始！",
                LogLevelType.INFO.getType(),
                EXECUTING,
                jobContext.getProgress());
        JobResult jobResult = new JobResult(SUCCESS, "创建索引成功");
        AbstractJdbcCatalog sinkCatalog = jobContext.getSinkCatalog();

        QualifiersModel sourceQualifiersModel = getSourceQualifiersModel();
        QualifiersModel sinkQualifiersModel = getSinkQualifiersModel();

        for (CatalogTable catalogTable: jobContext.getSourcetableMap().values()) {
            TablePath sourceTablePath = catalogTable.getTablePath();


            // 若限定了要处理的表名称
            if (job.getMissions() != null && !job.getMissions().isEmpty()) {
                if (TableNameComparator.containsTable(job.getMissions(), sourceTablePath.getTableName())) {
                    jobContext.addProgress(DbObjectType.INDEX.getTypeName(), job.getMissions().size());
                    JobResultItem jobResultItem = process(sinkCatalog, sourceTablePath, catalogTable, sourceQualifiersModel, sinkQualifiersModel);
                    jobResult.addItem(jobResultItem);
                }
            } else { // 若没有限定表，则是处理所有表
                jobContext.addProgress(DbObjectType.INDEX.getTypeName(), jobContext.getSourcetableMap().size());
                JobResultItem jobResultItem = process(sinkCatalog, sourceTablePath, catalogTable, sourceQualifiersModel, sinkQualifiersModel);
                jobResult.addItem(jobResultItem);
            }

        }

        if (jobResult.getFailtures() > 0) {
            jobResult.setCode(CREATE_TABLE_ERROR);
            jobResult.setMessage("创建索引失败");
            log.error("Create Some Indexes Of {} Failed.", jobContext.getValidSinkDatabase());
        } else {
            log.info("Create All Indexes Of {} Successfully.", jobContext.getValidSinkDatabase());
        }
        PostLog.sendLogModel(
                job.getJobId(),
                job.getTask().getTaskType().getName(),
                "创建索引结束！",
                LogLevelType.INFO.getType(),
                EXECUTING,
                jobContext.getProgress());
        return jobResult;
    }

private JobResultItem process(Catalog catalog, TablePath sourceTablePath, CatalogTable catalogTable, QualifiersModel sourceQualifiersModel, QualifiersModel sinkQualifiersModel) {
    String startTime = DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss.SSS");
        String sinkCatalogName = jobContext.getTaskConf().getSinkConf().getCatalogName();
        String sinkDatabaseName = jobContext.getTaskConf().getSinkConf().getDatabaseName();
        String sinkSchemaName = jobContext.getTaskConf().getSinkConf().getSchemaName();
        TablePath sinkTablePath = TablePath.of(sinkDatabaseName,
                sinkSchemaName,
                sourceTablePath.getTableName(),
                sinkQualifiersModel);
        TableIdentifier tableIdentifier = TableIdentifier.of(sinkCatalogName,
                sinkDatabaseName,
                sinkSchemaName,
                sourceTablePath.getTableName(),
                sinkQualifiersModel);
        CatalogTable sinkCatalogTable = CatalogTable.of(tableIdentifier,
                catalogTable.getTableSchema(),
                catalogTable.getOptions(),
                catalogTable.getComment(),
                sinkCatalogName);
        String sinkDDL = null;
        try {
            catalog.createIndex(sinkTablePath, sinkCatalogTable, false, jobContext.getTaskConf());
            List<String> indexSqls = catalog.getCreateIndexSqls(sinkTablePath, sinkCatalogTable, jobContext.getTaskConf());
            if (!CollectionUtils.isEmpty(indexSqls)) {
                sinkDDL = String.join(" \n", indexSqls);
            }
            log.info("Table {} index created!", tableIdentifier.toString());
        } catch (Exception e) {
            e.printStackTrace();
            PostLog.sendLogModel(
                    job.getJobId(),
                    job.getTask().getTaskType().getName(),
                    "创建表" + sourceTablePath.getTableName() + "的索引失败",
                    LogLevelType.ERROR.getType(),
                    EXECUTING,
                    jobContext.getProgress());
            return new JobResultItem(
                    sinkTablePath.getTableName(),
                    sinkTablePath.getCaseTableName(),
                    CREATE_INDEX_ERROR,
                    "创建表" + sourceTablePath.getTableName() + "的索引失败",
                    ExceptionUtils.toString(e),
                    DbObjectType.INDEX.getTypeName(),
                    startTime,
                    DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss.SSS"),
                    null,
                    sinkDDL);

        }
        PostLog.sendLogModel(
                job.getJobId(),
                job.getTask().getTaskType().getName(),
                "创建表" + sourceTablePath.getTableName() + "的索引成功",
                LogLevelType.INFO.getType(),
                EXECUTING,
                jobContext.getProgress());
        return new JobResultItem(
                sinkTablePath.getTableName(),
                sinkTablePath.getCaseTableName(),
                SUCCESS,
                "创建表" + sourceTablePath.getTableName() + "的索引成功",
                null,
                DbObjectType.INDEX.getTypeName(),
                startTime,
                DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss.SSS"),
                null,
                sinkDDL);
    }
}
